03

2023-05-29 07:22| 来源: 网络整理| 查看: 265

上一节，探讨了接口编程，这节将使用接口编程思想来实现项目中的dbrepo包所用功能。功能包括：

如何提取接口、优雅处理错误、基础CRDU、database批量添加数据、事务管理和实现

这节的核心思想就是 Accept interfaces, return structs(接受接口，返回结构体)；这是Go在绝大部场景都适用的一种模式。

废话不多说，上链接～

代码仓库在这👇👇👇重要的事情说三遍 EBook 代码仓库记得给个Star😘😘😘 代码仓库在这👇👇👇重要的事情说三遍 EBook 代码仓库记得给个Star😍😍😍

这就是本节创建文件：

首先创建 models/book.go package models import "time" type Book struct { ID uint `json:"id"` ISBN string `json:"isbn"` Title string `json:"title"` Poster string `json:"poster"` Pages uint `json:"pages"` Price float32 `json:"price"` PublishedAt time.Time `json:"published_at"` CreatedAt time.Time `json:"-"` UpdatedAt time.Time `json:"-"` }

现在解说一下 dbrepo文件结构和作用

├── base.go 基础/公共方法，可以嵌套在 xxxrepo.go 中，如：bookrepo.go ├── bookrepo.go book 表的crud/事务等操作 ├── error.go 定义此包的"能处理"的错误 ├── filter.go 过滤条件，如果分页查询 └── repository.go 整个包的入口，整合所有 xxxrepo.go

现在我希望要做的是，当用户想使用整个数据库操作句柄，可以通过 repository.go 中的 NewRepository获取到；如果仅仅想要bookrepo.go的操作句柄则通过NewBookRepo获取，这样可以很方便地拿到自己需要的。

创建 dbrepo/repository.go，写入： package dbrepo import ( "context" "database/sql" ) type Repository struct { Book BookRepo } // Queryable 提取 sql.DB 和 sql.Tx 公共的方法当作一个接口 type Queryable interface { Exec(query string, args ...any) (sql.Result, error) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) Query(query string, args ...any) (*sql.Rows, error) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) QueryRow(query string, args ...any) *sql.Row QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row Prepare(query string) (*sql.Stmt, error) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) } // NOTE: 采用这这种方式，是为了在执行事务上可以复用代码，更加灵活行 // NewRepository 创建一个Repository仓库，使用 Queryable 接口，同时兼容 sql.DB 和 sql.Tx 接口 func NewRepository(db Queryable) *Repository { return &Repository{ Book: NewBookRepo(db), } }

上面代码其中 BookRepo 是接口类型，稍后定义，现在了解为什么要抽取Queryable这么一个接口；这是因为 database/sql 标准库在执行基础的CRUD中使用的是sql.DB而在执行事务中使用的是sql.Tx，这就意味着，我一旦直接使用了sql.DB去编程基础的CRUD代码后，在执行事务操作时则无法利用这些CRUD代码，这很有可能就要多维护一份重复代码。因此才提取 sql.DB 和 sql.Tx 公共的方法当作一个接口。然后在func NewRepository(db Queryable) *Repository中既可以传入sql.DB又可以传入sql.Tx。此时当我在编写事务时，使用sql.Tx去New一个Repository就达到共享代码好处。

接下来看一下 dbrepo/bookrepo.go 上半部分代码： package dbrepo import ( "context" "fmt" "strings" "time" "github.com/lightsaid/ebook/internal/models" "github.com/lightsaid/ebook/pkg/logger" ) type BookRepo interface { Insert(b *models.Book) (uint, error) Get(id uint) (*models.Book, error) List(page Pager) ([]*models.Book, error) Update(id uint, book *models.Book) error Del(id uint) error BulkInsert(books []*models.Book) (int64, error) TestTx(id uint) error } var _ BookRepo = (*bookRepo)(nil) type bookRepo struct { dbBase DB Queryable } func NewBookRepo(db Queryable) *bookRepo { return &bookRepo{ DB: db, } }

首先是定义BookRepo接口，有哪些方法允许被外部使用，接着定义bookRepo结构体去实现BookRepo接口，而dbBase是定义在base.go中的struct，可以先忽略，后续在讲。

一定要进行接口检查 var _ BookRepo = (*bookRepo)(nil)，bookRepo 就是为了实现 BookRepo的，如果没有实现，vs code 就会直接爆红，编译不通过，这是接受接口，返回结构体提前感知有没有错误的好方法。

在正式编写基础CRUD之前，先了解一下database/sql库常用方法。

database/sql 常见方法 * QueryRow() 查询单行数据 * QueryRowContext() * Query() 查询多行数据 * QueryContext() * Exec() 执行sql,不返回任何行，用户insert/update/delete * ExecContext() 查看它们定义 func (db *DB) QueryRow(query string, args ...any) *Row { return db.QueryRowContext(context.Background(), query, args...) } func (db *DB) Query(query string, args ...any) (*Rows, error) { return db.QueryContext(context.Background(), query, args...) } func (db *DB) Exec(query string, args ...any) (Result, error) { return db.ExecContext(context.Background(), query, args...) }

查看它们的定义可以得知，其实它们最终都是调用xxxContext()方法。上下文是一个很重要参数，建议不要忽略，在真实生产环境中都将以使用超时上下，避免资源被耗尽。上面哪些方法都是常用的操作，但是我还是建议使用预查询PrepareContext 提高性能。

下面来看看 bookdbrepo.go 全部代码实现，写得非常细致。 package dbrepo import ( "context" "fmt" "strings" "time" "github.com/lightsaid/ebook/internal/models" "github.com/lightsaid/ebook/pkg/logger" ) type BookRepo interface { Insert(b *models.Book) (uint, error) Get(id uint) (*models.Book, error) List(page Pager) ([]*models.Book, error) Update(id uint, book *models.Book) error Del(id uint) error BulkInsert(books []*models.Book) (int64, error) TestTx(id uint) error } var _ BookRepo = (*bookRepo)(nil) type bookRepo struct { dbBase DB Queryable } func NewBookRepo(db Queryable) *bookRepo { return &bookRepo{ DB: db, } } func (b *bookRepo) Insert(book *models.Book) (uint, error) { insertSQL := `insert into book( isbn, title, poster, pages, price, published_at )values( ?,?,?,?,?,? )` // ret, err := b.DB.Exec(insertSQL, book.ISBN, book.Title, book.Poster, book.Pages, book.Price, book.PublishedAt) // if err != nil { // return 0, err // } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() stmt, err := b.DB.PrepareContext(ctx, insertSQL) if err != nil { return 0, err } defer stmt.Close() ret, err := stmt.Exec(book.ISBN, book.Title, book.Poster, book.Pages, book.Price, book.PublishedAt) if err != nil { return 0, handleMySQLError(err) } insertID, err := ret.LastInsertId() if err != nil { return 0, handleMySQLError(err) } return uint(insertID), nil } func (b *bookRepo) Get(id uint) (*models.Book, error) { querySQL := `select id, isbn, title, poster, pages, price, published_at from book where id = ?` ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() stmt, err := b.DB.PrepareContext(ctx, querySQL) if err != nil { return nil, err } defer stmt.Close() var book = new(models.Book) err = stmt.QueryRow(id).Scan(&book.ID, &book.ISBN, &book.Title, &book.Poster, &book.Pages, &book.Price, &book.PublishedAt) return book, handleMySQLError(err) } func (b *bookRepo) List(page Pager) ([]*models.Book, error) { querySQL := `select id, isbn, title, poster, pages, price, published_at from book limit ? offset ?` ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() stmt, err := b.DB.PrepareContext(ctx, querySQL) if err != nil { return nil, err } defer stmt.Close() rows, err := stmt.Query(page.Limit(), page.Offset()) if err != nil { return nil, err } defer rows.Close() var books = make([]*models.Book, 0) for rows.Next() { var book = new(models.Book) err = rows.Scan(&book.ID, &book.ISBN, &book.Title, &book.Poster, &book.Pages, &book.Price, &book.PublishedAt) if err != nil { return nil, err } books = append(books, book) } if rows.Err() != nil { return nil, rows.Err() } return books, nil } func (b *bookRepo) Update(id uint, book *models.Book) error { updateSQL := `update book set isbn = ?, title = ?, poster = ?, pages = ?, price = ?, published_at = ?, updated_at = now() where id = ? ` qBook, err := b.Get(id) if err != nil { return handleMySQLError(err) } if book.ISBN != "" { qBook.ISBN = book.ISBN } if book.Title != "" { qBook.Title = book.Title } if book.Poster != "" { qBook.Poster = book.Poster } if book.Pages != 0 { qBook.Pages = book.Pages } if book.Price >= 0 { qBook.Price = book.Price } if !book.PublishedAt.IsZero() { qBook.PublishedAt = book.PublishedAt } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() stmt, err := b.DB.PrepareContext(ctx, updateSQL) if err != nil { return err } defer stmt.Close() ret, err := stmt.Exec( qBook.ISBN, qBook.Title, qBook.Poster, qBook.Pages, qBook.Price, qBook.PublishedAt, qBook.ID, ) if err != nil { return err } aff, err := ret.RowsAffected() if err != nil { return fmt.Errorf("%w:%v", ErrUpdateFailed, err) } // 最终更新成功 if aff > 0 { return nil } return ErrUpdateFailed } func (b *bookRepo) Del(id uint) error { delSQL := `delete from book where id = ?` ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() stmt, err := b.DB.PrepareContext(ctx, delSQL) if err != nil { return err } defer stmt.Close() ret, err := stmt.Exec(id) if err != nil { return handleMySQLError(err) } aff, err := ret.RowsAffected() if err != nil { return err } if aff > 0 { return nil } // SQL 执行成功，但是没有影响行数，那就是数据已经不存在了 return ErrNotFound } // BulkInsert 批量插入, 并没有对数量做限制，因此需要注意插入数据的数量，以防超出SQL语句的大小限制 func (b *bookRepo) BulkInsert(books []*models.Book) (int64, error) { // 批量插入，采用 insert into tbName(field1, field2) values(?, ?),values(?,?)... insertSQL := `insert into book( isbn, title, poster, pages, price, published_at ) values ` var values []string var params []interface{} for _, book := range books { values = append(values, "(?,?,?,?,?,?)") params = append(params, book.ISBN, book.Title, book.Poster, book.Pages, book.Price, book.PublishedAt) } // 拼接 values => "insert into ... values (?,?...),(?,?...)" insertSQL += strings.Join(values, ",") logger.InfoLog.Println("book BulkInsert sql: ", insertSQL) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() stmt, err := b.DB.PrepareContext(ctx, insertSQL) if err != nil { return 0, err } defer stmt.Close() ret, err := stmt.Exec(params...) if err != nil { return 0, handleMySQLError(err) } aff, err := ret.RowsAffected() if err != nil { return 0, err } return aff, nil } // TestTx NOTE: 测试事物，获取图书和更新图书，不会修改任何数据 func (b *bookRepo) TestTx(id uint) error { err := b.dbBase.execTx(context.Background(), b.DB, func(r *Repository) error { // 获取 book, err := r.Book.Get(id) if err != nil { return err } // 更新 return r.Book.Update(id, book) }) return err }

在这里先讲解 handleMySQLError处理DB错误函数，它定义在 error.go中

error.go package dbrepo import ( "database/sql" "errors" "strings" "github.com/go-sql-driver/mysql" ) var ( ErrNotFound = errors.New("记录不存在") ErrISBNAlreadyExists = errors.New("isbn 已经存在") ErrUpdateFailed = errors.New("更新失败") ErrDelFailed = errors.New("删除失败") ) func IsCustomDBError(err error) bool { switch { case errors.Is(err, ErrNotFound), errors.Is(err, ErrISBNAlreadyExists), errors.Is(err, ErrUpdateFailed), errors.Is(err, ErrDelFailed): return true } return false } // 统一处理mysql错误 func handleMySQLError(err error) error { if err != nil { if err == sql.ErrNoRows { return ErrNotFound } var mysqlErr *mysql.MySQLError // 重复键错误, 具体哪个字段重复，由具体业务判断 if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { // NOTE: unq_isbn 是数据库定义唯一索引名字 if strings.Contains(err.Error(), "unq_isbn") { return ErrISBNAlreadyExists } // NOTE: 根据业务添加其他的 } } return err } 重点说一下 handleMySQLError 函数和Go语言如何处理错误。

handleMySQLError 函数是一个集中处理db 错误函数，为什么有的地方使用，有的地方不使用handleMySQLError处理错误呢？

在Go语言中，如果发生错误，首先就是判断能不能处理，能处理则处理，不能处理则往调用抛出去。

首先当API服务在客户端调用时（如：前端），如果请求不正确，发生错误，应该有友好提示或者引导的作用。而不应该就一些数据库内部的错误直接返回给客户。开发者应该假设客户不懂技术，乱提示非常令人疑惑；再者直接将数据库的错误显示给客户反而增加系统被攻击的风险。因为这些错误信息可能会暴露系统一些重要信息。

因此我在error.go中定义了我能感知的错误类型，可以显示给客户看到提示，当handlers层调用dbrepo层时，如果错误被处理了，我们可以选择不再错误，直接返回响应；对于一些不确定错误，输出日志，记录参数，返回500之类的状态码。

同时也定义了IsCustomDBError函数，判断错误类型以供调用方使用。

Pager 结构体,用于过滤和分页定义在 filter.go 中，这里目前还简单不用多说。

package dbrepo type Pager struct { size int // 每页多条 page int // 当前第几页 } // NewPager 创建一个分页器 page >= 1, "insert into ... values (?,?...),(?,?...)" insertSQL += strings.Join(values, ",")

最后要说的是事务操作, 先看看base.go定义的基础方法，execTx 方法就是一个执行事务基础方法，它主要是处理事务开启、回滚/提交。因此在编写错误的时候，不用过多关心基础步骤，主要把核心逻辑处理好即可。

package dbrepo import ( "context" "database/sql" "errors" "fmt" ) // dbBase 内部实现通用方法，嵌套在repo，提供给repo使用 type dbBase struct{} // execTx 定义个执行事务公共的方法 func (dbBase) execTx(ctx context.Context, qb Queryable, fn func(*Repository) error) error { // qb => sql.DB/sql.Tx db, ok := qb.(*sql.DB) if !ok { return errors.New("b.DB not is *sql.DB") } // 开启事务 tx, err := db.BeginTx(ctx, nil) if err != nil { return err } r := NewRepository(tx) if _, err = r.Book.Get(6); err != nil { return err } q := NewRepository(tx) if err = fn(q); err != nil { // 执行不成功，则 Rollback if rbErr := tx.Rollback(); rbErr != nil { return fmt.Errorf("tx err: %v, rb err: %v", err, rbErr) } return err } // 执行成功 Commit return tx.Commit() }

最后在dbrepo.go中的 TestTx 方法中测试事务操作。

代码仓库在这👇👇👇重要的事情说三遍，持续更新，欢迎Star EBook 代码仓库记得给个Star😘😘😘 代码仓库在这👇👇👇重要的事情说三遍，持续更新，欢迎Star EBook 代码仓库记得给个Star😍😍😍

【本文地址】

公司简介

联系我们